package com.trendytech.aidata.udf.sql_function;

import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
import java.util.Objects;

/**
 * TwoColumSplitFunction 是一个自定义的 Flink 表函数，
 * 用于将输入的字符串按逗号分割成两列，并将分割结果封装成 Row 对象输出。
 */
@FunctionHint(output = @DataTypeHint("ROW<f0 STRING,f1 STRING>"))
public class TwoColumSplitFunction extends TableFunction<Row> {

    /**
     * eval 方法用于处理输入的字符串，将其按逗号分割成两列并输出。
     *
     * @param str 输入的待分割字符串
     */
    public void eval(String str) {
        // 检查输入的字符串是否为 null
        if (Objects.isNull(str)) {
            // 如果为 null，收集一个包含两个空字符串的 Row 对象
            collect(Row.of("", ""));
        } else {
            // 使用 split 方法按逗号分割字符串，limit 为 -1 表示保留末尾的空字符串
            String[] split = str.split(",", -1);
            // 根据分割结果的长度进行不同处理
            if (split.length == 1) {
                // 如果分割结果只有一个元素，收集一个 Row 对象，第一列是该元素，第二列是空字符串
                collect(Row.of(split[0], ""));
            } else if (split.length == 2) {
                // 如果分割结果有两个元素，收集一个 Row 对象，包含这两个元素
                collect(Row.of(split[0], split[1]));
            } else {
                // 如果分割结果长度不是 1 也不是 2，收集一个包含两个空字符串的 Row 对象
                collect(Row.of("", ""));
            }
        }
    }
}